package com.atguigu.flink.streamapi.sink;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * Created by Smexy on 2023/1/30
 *

 CREATE TABLE  watersensor(
 `id` String,
 ts UInt64,
 vc UInt32

 )
 ENGINE = ReplacingMergeTree(ts)
 ORDER BY id;
 */
public class Demo5_JDBCSink
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
            以写入ck为例。
                保存每种传感器最新的vc

            sink(
            String sql, //写入的sql语句
            JdbcStatementBuilder<T> statementBuilder // 用来填充ps的占位符
            JdbcExecutionOptions executionOptions, //写入其他参数
            JdbcConnectionOptions connectionOptions //创建链接的参数
            )
         */
        SinkFunction<WaterSensor> jdbcSink = JdbcSink.sink(
            "insert into watersensor values(?,?,?)",
            new JdbcStatementBuilder<WaterSensor>()
            {
                //收到一条WaterSensor，如何去填充占位符
                @Override
                public void accept(PreparedStatement ps, WaterSensor value) throws SQLException {
                    ps.setString(1, value.getId());
                    ps.setLong(2, value.getTs());
                    ps.setInt(3, value.getVc());
                }
            },
            JdbcExecutionOptions.builder()
                                .withMaxRetries(3)  //写失败后，最多重试多少次
                                .withBatchSize(100)  //一批最多存放多少条
                                .withBatchIntervalMs(3000) //当前不满足一批数据，固定的时间间隔写出
                                .build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl("jdbc:clickhouse://hadoop102:8123/default")
                .build()
        );

        env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .addSink(jdbcSink);


                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
